package com.camemax.operators;

import com.camemax.pojo.LoginEvent;
import com.camemax.pojo.LoginFailedWarning;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.shaded.guava18.com.google.common.collect.Iterators;
import org.apache.flink.shaded.guava18.com.google.common.collect.Lists;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.Iterator;

/*
* 实现登录监控状态：
*    1. 当登录失败时， 赋予数据流监测状态， 记录当前数据流（LoginEvent）入列（ListState）
* */
public class LoginFailedProcess extends KeyedProcessFunction<Long, LoginEvent, LoginFailedWarning> {

    private ListState<LoginEvent> loginFailedEvents;
    private ValueState<Long> failedTime;
    private Integer loginFailedThreshold;

    public LoginFailedProcess(Integer threshold){
        this.loginFailedThreshold = threshold;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        loginFailedEvents = getRuntimeContext().getListState(
                new ListStateDescriptor<>("failed-login-history", LoginEvent.class)
        );

        failedTime = getRuntimeContext().getState(
                new ValueStateDescriptor<Long>("failed-time", Long.class)
        );
    }

    @Override
    public void processElement(LoginEvent value, Context ctx, Collector<LoginFailedWarning> out) throws Exception {
        if ("fail".equals(value.getLoginStatus())){

            if( failedTime.value() == null ){
                long initiateFailedTime = value.getLoginDate() * 1000L ;
                long trulyFailedTime = initiateFailedTime + 2000L ;
                ctx.timerService().registerEventTimeTimer(trulyFailedTime);
                failedTime.update(trulyFailedTime);
            }

            loginFailedEvents.add(value);
        }else{
            if ( failedTime.value() != null ){
                long trulyFailedTime = failedTime.value();
                ctx.timerService().deleteProcessingTimeTimer(trulyFailedTime);
            }
            loginFailedEvents.clear();
            failedTime.clear();
        }
    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<LoginFailedWarning> out) throws Exception {

        ArrayList<LoginEvent> loginFailEvents = Lists.newArrayList(loginFailedEvents.get());
        if ( loginFailEvents.size() >= loginFailedThreshold ){
            out.collect(
                    new LoginFailedWarning(
                            ctx.getCurrentKey()
                            , "Login Failed Warning!"
                            , LocalDateTime.ofEpochSecond( loginFailEvents.get(0).getLoginDate()
                                , 0
                                , ZoneOffset.ofHours(8))
                                .format(DateTimeFormatter.ofPattern("yyyy/MM/dd hh:mm:ss")
                                )
                            , LocalDateTime.ofEpochSecond( loginFailEvents.get(loginFailEvents.size() - 1).getLoginDate()
                                , 0
                                , ZoneOffset.ofHours(8))
                                .format(DateTimeFormatter.ofPattern("yyyy/MM/dd hh:mm:ss")
                                )
                    )
            );
        }
    }
}

class MyLoginFailedProcess extends KeyedProcessFunction<Long, LoginEvent, LoginFailedWarning>{
    private ListState<LoginFailedWarning> warningRecord;
    private ValueState<String> firstFailed;
    private ValueState<String> lastFailed;

    @Override
    public void open(Configuration parameters) {
        warningRecord = getRuntimeContext().getListState(
                new ListStateDescriptor<>("failed-login-history", LoginFailedWarning.class)
        );

        firstFailed = getRuntimeContext().getState(
                new ValueStateDescriptor<>("first-failed-time", String.class)
        );

        lastFailed = getRuntimeContext().getState(
                new ValueStateDescriptor<>("last-failed-time", String.class)
        );
    }

    // 当出现状态为failed时，执行以下操作：
    // 1. 将当前数据封装成LoginFailedWarning包装类
    // 2. 写入ListState，用于状态标识和之后的判断
    // 3. ctx.registry注册窗口触发时间
    @Override
    public void processElement(LoginEvent value, Context ctx, Collector<LoginFailedWarning> out) throws Exception {

        if ("fail".equals(value.getLoginStatus())) {
            // 1. 当第一次登录失败时
            if ( firstFailed.value() == null ){
                long firstFailedTime = value.getLoginDate() * 1000L;
                firstFailed.update(String.valueOf(firstFailedTime));
                long lastFailedTime = firstFailedTime + 2000L;
                lastFailed.update( String.valueOf(lastFailedTime));
                ctx.timerService().registerEventTimeTimer(lastFailedTime);
            }

            warningRecord.add(
                    new LoginFailedWarning(
                            ctx.getCurrentKey()
                            , "Failed Login Alarm!"
                            , null
                            , null
                    )
            );

        } else {
            // 清空时间窗口
            if ( firstFailed.value() != null ){
                long getWindowEnd = Long.parseLong(lastFailed.value());
                ctx.timerService().deleteProcessingTimeTimer(getWindowEnd);
            }
            warningRecord.clear();
            firstFailed.clear();
            lastFailed.clear();
        }
    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<LoginFailedWarning> out) throws Exception {
        Iterator<LoginFailedWarning> iterator = warningRecord.get().iterator();
        int failedCount = Iterators.size(iterator);
        if(failedCount >= 3) {
            long firstFailedTime = Long.parseLong(firstFailed.value());
            long lastFailedTime = Long.parseLong(lastFailed.value());
            String firstFailed = LocalDateTime.ofEpochSecond(firstFailedTime / 1000
                    , 0
                    , ZoneOffset.ofHours(8))
                    .format(DateTimeFormatter.ofPattern("yyyy/MM/dd hh:mm:ss"));

            String lastFailed = LocalDateTime.ofEpochSecond(lastFailedTime / 1000
                    , 0
                    , ZoneOffset.ofHours(8))
                    .format(DateTimeFormatter.ofPattern("yyyy/MM/dd hh:mm:ss"));

            out.collect(
                    new LoginFailedWarning(
                            ctx.getCurrentKey()
                            , " Failed Login Warning!"
                            , firstFailed
                            , lastFailed
                    )
            );
        }
        warningRecord.clear();
        firstFailed.clear();
        lastFailed.clear();
    }
}